#ifndef __M_MSG_H__
#define __M_MSG_H__

#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <list>

namespace mq
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"

    using Message = mq::MQMessage;
    using MessagePtr = std::shared_ptr<mq::Message>;

    // 消息文件管理类。
    class MessageMapper
    {
    public:
        using Ptr = std::shared_ptr<MessageMapper>;

        MessageMapper( std::string &basedir, const std::string &filename)
        {
            // 如果basedir为const则不能向basedir里面推送数据。
            if (basedir.size() == 0)
            {
                basedir += "./";
            }
            if (basedir.back() != '/')
            {
                basedir += "/";
            }
            _datafile = basedir + filename + DATAFILE_SUBFIX;
            _tmpfile = basedir + filename + TMPFILE_SUBFIX;
            // 没有文件夹就创建。
            // 创建不了就报错。
            if (mq::Helper::FileHelper(basedir).exist() == false)
            {
                assert(mq::Helper::FileHelper::createDirectory(basedir));
            }

            createMsgfile();
        }
        bool createMsgfile()
        {
            if (mq::Helper::FileHelper(_datafile).exist() == true)
            {
                return true;
            }

            bool ret = mq::Helper::FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建datafile失败！");

                return false;
            }

            return true;
        }
        void removeMsgFile()
        {
            mq::Helper::FileHelper::removeFile(_datafile);
            mq::Helper::FileHelper::removeFile(_tmpfile);

            return ;
        }
        bool insert(MessagePtr& msg)
        {
            // 不是将文件删除，而是将有效标志位修改为零。
            // 然后修改文件数据
            
            return insert(_datafile,msg);
        }
        bool remove(MessagePtr& msg)
        {
            // 将有效标志位修改为零。
            msg->mutable_payload()->set_valid("0");
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                DLOG("不能修改源文件的数据，因为内容长度不匹配。");

                return false;
            }
            mq::Helper::FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                // 删除队列失败。
                DLOG("向数据文件写入失败");

                return false;
            }

            DLOG("确认消息后，删除持久化消息成功：%s", msg->payload().body().c_str());

            return true;
        }
        // 将源文件删除。临时文件替换为源文件。
        std::list<MessagePtr> gc()
        {
            std::list<MessagePtr> result;
            bool ret = load(result);
            if (ret == false)
            {
                DLOG("加载有效数据失败！");

                return result;
            }
            
            // DLOG("垃圾回收，得到有效消息数量：%d", result.size());
            // 2. 将有效数据，进行序列化存储到临时文件中
            mq::Helper::FileHelper::createFile(_tmpfile);
            // 向_tmpfile文件中写入。
            for (auto &msg : result)
            {
                DLOG("向_tmpfile文件中写入：%ld",msg->payload().body().size());
                bool ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    DLOG("向临时文件写入失败。")

                    return result;
                }
            }
            DLOG("垃圾回收后，向临时文件写入数据完毕，临时文件大小: %ld",mq::Helper::FileHelper(_tmpfile).size());

            // 删除源文件。
            ret = mq::Helper::FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                DLOG("删除源文件失败。");
                return result;
            }

            // 修改临时文件名，为源文件名称
            mq::Helper::FileHelper helper(_tmpfile);
            ret = helper.rename(_tmpfile, _datafile);
            ret = helper.rename(_datafile);
            if (ret == false)
            {
                DLOG("修改文件失败。");
                return result;
            }

            return result;
        }

    private:
        bool load(std::list<MessagePtr>& result)
        {
            // 加载出文件中所有的有效数据； 
            // 存储格式 4 字节长度| 数据 | 4 字节长度 | 数据.....
            // 序列化数据。
            mq::Helper::FileHelper helper(_datafile);
            size_t offset = 0 , msg_size = 0;
            size_t fsize = helper.size();
            bool ret;
            while(offset < fsize) 
            {
                ret = helper.read((char*)&msg_size,offset,sizeof(msg_size));
                if(ret == false)
                {
                    DLOG("读取文件失败。");

                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msg_size,'\0');
                ret = helper.read(&msg_body[0],offset,msg_size);
                if(ret == false)
                {
                    DLOG("读取消息失败。");

                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<MQMessage>();
                // 反序列化的是payload而不是message类。
                msgp->mutable_payload()->ParseFromString(msg_body);
                DLOG("加载到有效数据：%s", msgp->payload().body().c_str()) ;
                // 如果是无效消息，直接跳过下一个。
                if(msgp->payload().valid() == "0")
                {   
                    DLOG("加载无效消息%s",msgp->payload().body().c_str());
                    continue;
                }
                // 有效消息保存起来。
                result.push_back(msgp);
            }

            return true;
        }
        bool insert(const std::string& filename,MessagePtr& msg)
        {
            bool ret;
            // 新增数据都是添加在文件末尾的 
            // 1. 进行消息的序列化，获取到格式化后的消息
            std::string body = msg->payload().SerializeAsString();
            // 2. 获取文件长度;
            mq::Helper::FileHelper helper(filename);
            size_t offset = helper.size();
            size_t msg_size = body.size();
            // 3. 将数据写入文件的指定位置
            //写入逻辑：1. 先写入 4 字节数据长度， 2， 再写入指定长度数据 
            ret = helper.write((char*)&msg_size,offset,sizeof(msg_size));
            // offset += sizeof(msg_size);
            if(ret == false)
            {
                DLOG("将文本大小写入文件失败。");
                return false;
            }
            ret = helper.write(&body[0],offset + sizeof(msg_size),msg_size);
            if(ret == false)
            {
                DLOG("将文本写入文件失败。");
                return false;
            }

            // 4. 更新 msg 中的实际存储信息 
            msg->set_offset(offset + sizeof(size_t));
            msg->set_length(body.size()); 

            return true;
        }
    private:
        std::string _datafile;
        std::string _tmpfile;
    };
    // 队列消息。
    class QueueMessage
    {
    public:
        using Ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string& basedir,const std::string& qname)
            :_mapper(basedir,qname)
            ,_qname(qname)
            ,_valid_count(0)
            ,_total_count(0)
        {}
        std::string Qname()
        {
            
            return _qname;
        }
        void recovery()
        {
            // 恢复历史消息。
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();

            for(auto& msg : _msgs)
            {
                DLOG("插入消息:%s",msg->payload().body().c_str());
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
            }
            _total_count = _valid_count = _msgs.size();

            return ;
        }
        bool insert(const BasicProperties* bp,const std::string& body,bool queue_is_durable)
        {
            DLOG("msg insert");

            //1. 构造消息对象。
            MessagePtr msg = std::make_shared<MQMessage>();
            // bp不为空，内含有有效数据。
            if(bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable == true ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                // 是否设置持久化模式
                msg->mutable_payload()->mutable_properties()->
                set_delivery_mode(mode);

                // 设置routing_key。
                msg->mutable_payload()->mutable_properties()->
                set_rounting_key(bp->rounting_key());

                msg->mutable_payload()->mutable_properties()->
                set_id(bp->id());

                msg->mutable_payload()->
                set_body(body);
            }
            // bp为空，内部没有有效数据，根据函数的其他参数来进行判断。
            else
            {
                // 二进制bool值进行判断
                DeliveryMode mode = queue_is_durable == true ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                
                // 是否设置持久化模式
                // mutable返回的指针才能修改值，常量不需要，直接返回值。
                msg->mutable_payload()->mutable_properties()->
                set_delivery_mode(mode);
                
                // 设置routing_key。
                msg->mutable_payload()->mutable_properties()->
                set_rounting_key("");
                
                msg->mutable_payload()->mutable_properties()->
                set_id(mq::Helper::UuidHelper::uuid());
                
                msg->mutable_payload()->
                set_body(body);
            }
            // 访问公共资源。
            std::unique_lock<std::mutex> lock(_mutex);

            // 判断消息是否需要持久化。
            if(msg->payload().properties().delivery_mode() == mq::DeliveryMode::DURABLE)
            {
                // 在持久化存储中表示数据有效。
                msg->mutable_payload()->set_valid("1");
                // 
                bool ret = _mapper.insert(msg);
                if(ret == false)
                {
                    DLOG("持久化存储消息：%s失败了！",body.c_str());

                    return false;
                }
                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
            }

            // 内存的管理。
            _msgs.push_back(msg);

            DLOG("insert successful");

            return true;
        }
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 
            if(_msgs.size() == 0)
            {
                DLOG("该队列没有消息");

                return MessagePtr();
            }
            // 从_msgs的首部取出消息来。
            MessagePtr msg = _msgs.front();
            // 插入到待确认队列中去。
            // 一旦确认就直接删除。
            _msgs.pop_front();
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));

            return msg;
        }
        // 每次删除消息后，判断是否需要垃圾回收。
        bool remove(const std::string& msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1.从待确认消息队列中查找消息。
            auto it = _waitack_msgs.find(msg_id);
            if(it == _waitack_msgs.end())
            {
                DLOG("没有要删除的消息：%s！",msg_id.c_str());
                return true;
            }
            // 2.存在该条消息，从队列中删除。
            if(it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3.删除持久化消息。
                _mapper.remove(it->second); 
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                gc();
                // 内部判断是否需要回收，需要的话就回收一下。
            }
            // 4.删除内存中的消息。
            DLOG("确认消息后，删除消息的管理成功：%s",it->second->payload().body().c_str()); 
            _waitack_msgs.erase(msg_id); // 删除前打印，而不是删除后释放了，才打印。
            
            return true;
        }
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            
            return _msgs.size();
        }
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            
            return _total_count;
        }
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);

            return _durable_msgs.size();
        }
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            
            return _waitack_msgs.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }
    private:
        bool GCCheck()
        {
            // 持久化的消息总量大于 2000， 且其中有效比例低于 50%则需要持久化 
            if(_total_count >= 2000 && _valid_count * 10 / _total_count < 50)
            {
                return true;
            }

            return false;
        }
        void gc()
        {
            //1. 进行垃圾回收，获取到垃圾回收后，有效的消息信息链表
            if(GCCheck() == false)
            {
                return ;
            }
            std::list<MessagePtr> msgs = _mapper.gc();
            for(auto& msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    // 未进行管理。
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                    // 重新插入。
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                    continue;
                }
                //2. 更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            //3. 更新当前的有效消息数量 & 总的持久化消息数量 
            _valid_count = _total_count = _msgs.size();
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count;
        size_t _total_count;
        MessageMapper _mapper;
        // 待推送消息
        std::list<MessagePtr> _msgs;
        // 持久化消息hash
        std::unordered_map<std::string,MessagePtr> _durable_msgs;
        // 待确认消息hash
        std::unordered_map<std::string,MessagePtr> _waitack_msgs;
    };
    class MessageManager
    {
    public:
        using Ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string& basedir)
            :_basedir(basedir)
        {}
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);

            for(auto& qmsg : _queue_msgs)
            {
                qmsg.second->clear();
            }
        }
        void initQueueMessage(const std::string& qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _queue_msgs.find(qname);
                if(it != _queue_msgs.end())
                {
                    // 已经存在
                    DLOG("初始化队列时，队列%s已存在。不用创建。",qname.c_str());
                    DLOG("it->qname:%s",it->second->Qname().c_str());

                    return ;
                }
                qmp = std::make_shared<QueueMessage>(_basedir,qname);
                _queue_msgs.insert(std::make_pair(qname,qmp));
            }
            qmp->recovery();
        }
        void destroyQueueMessage(const std::string& qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    // 不存在或者已经被删除了。

                    return ;
                }
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear();
        }
        bool insert(const std::string &qname, BasicProperties *bp,
            const std::string &body, bool queue_is_durable)
        {
            DLOG("insert qmp");
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("向队列%s 新增消息失败：没有找到消息管理句柄!",qname.c_str());

                    return false;
                }
                qmp = it->second;
            }

            return qmp->insert(bp,body,queue_is_durable);
        }
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s 队首消息失败：没有找到消息管理句柄!",qname.c_str());

                    return MessagePtr();
                }
                qmp = it->second;
            }

            return qmp->front();
        }
        bool ack(const std::string& qname, const std::string& msg_id)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("确认队列%s消息失败：没有找到消息管理",qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
            DLOG("确认队列%s消息成功",qname.c_str());
            
            return true;
        }
        size_t getable_count(const std::string& qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s 待推送消息数量失败：没有找到消息管理句柄!",qname.c_str());

                    return 0;
                }
                qmp = it->second;
            }

            return qmp->getable_count();
        }
        size_t total_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s 总持久化消息数量失败：没有找到消息管理句柄!",
                        qname.c_str());

                    return 0;
                }
                qmp = it->second;
            }

            return qmp->total_count();
        }
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s 总持久化消息数量失败：没有找到消息管理句柄!",
                        qname.c_str());

                    return 0;
                }
                qmp = it->second;
            }

            return qmp->durable_count();
        }
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::Ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s 总持久化消息数量失败：没有找到消息管理句柄!",
                        qname.c_str());
                        
                    return 0;
                }
                qmp = it->second;
            }

            return qmp->waitack_count();
        }
        
    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string,QueueMessage::Ptr> _queue_msgs;
    };
}

#endif